package com.atguigu.gamll.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gamll.realtime.beans.TableProcess;
import com.atguigu.gamll.realtime.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private OutputTag<JSONObject> dimTag;
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    private Connection conn;

    public TableProcessFunction(OutputTag<JSONObject> dimTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.dimTag = dimTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // 注册驱动
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        // 建立连接
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // 获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        // 获取业务数据库的表名
        String table = jsonObject.getString("table");
        // 获取业务数据库的操作类型
        String type = jsonObject.getString("type");
        // 如果使用Maxwell读取历史数据，操作类型是bootstrap-insert，需要进行修正
        if ("bootstrap-insert".equals(type)) {
            type = "insert";
            jsonObject.put("type", type);
        }

        // 拼接从状态中获取配置信息的key
        String key = table + ":" + type;
        // 根据key从广播状态中获取当前操作所对应的配置信息
        TableProcess tableProcess = broadcastState.get(key);
        if (tableProcess != null) {
            // 向下游传递前，都应该携带上要往哪个地方存储的属性
            String sinkTable = tableProcess.getSinkTable();
            jsonObject.put("sink_table", sinkTable);
            // 对不需要的属性进行过滤
            filterColumn(jsonObject.getJSONObject("data"), tableProcess.getSinkColumns());
            // 判断是事实还是维度
            if (TableProcess.SINK_TYPE_HBASE.equals(tableProcess.getSinkType())) {
                // 维度数据 --> 侧输出流
                ctx.output(dimTag, jsonObject);
            } else if (TableProcess.SINK_TYPE_KAFKA.equals(tableProcess.getSinkType())) {
                // 事实数据 --> 主流
                out.collect(jsonObject);
            }
        } else {
            System.out.println("No this key in tableProcess:" + key);
        }
    }

    // 处理广播流
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        // 将反序列化后的json字符串转换为json对象
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        // 获取配置表中的一条信息
        JSONObject dataObj = jsonObj.getJSONObject("data");
        // 将配置信息封装为一个TableProcess对象
        TableProcess tableProcess = dataObj.toJavaObject(TableProcess.class);
        // 获取业务数据库表名
        String sourceTable = tableProcess.getSourceTable();
        // 获取操作类型
        String operateType = tableProcess.getOperateType();
        // 获取输出类型  hbase kafka
        String sinkType = tableProcess.getSinkType();
        // 获取输出目的地
        String sinkTable = tableProcess.getSinkTable();
        // 获取建表字段---指定哪些字段要保留
        String sinkColumns = tableProcess.getSinkColumns();
        // 获取建表主键
        String sinkPk = tableProcess.getSinkPk();
        // 获取建表扩展
        String sinkExtend = tableProcess.getSinkExtend();

        // 拼接key
        String key = sourceTable + ":" + operateType;

        // 判断是不是维度数据，如果是维度数据，就把维度表创建出来
        if (TableProcess.SINK_TYPE_HBASE.equals(sinkType) && "insert".equals(operateType)) {
            checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);
        }

        // 获取广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        // 将配置信息放到广播状态中
        broadcastState.put(key, tableProcess);
    }

    /**
     * 字段过滤
     * @param dataJsonObj 要过滤的json对象
     * @param sinkColumns 保留的字段
     */
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        entrySet.removeIf(entry -> !columnList.contains(entry.getKey()));
    }

    // 创建维度表
    private void checkTable(String tableName, String fields, String pk, String ext) {
        if (pk == null) {
            pk = "id";
        }
        if (ext == null) {
            ext = "";
        }
        // 拼接建表语句
        StringBuilder createSql = new StringBuilder("create table if not exists " + GmallConfig.HBASE_SCHEMA + "." + tableName + " (");
        String[] fieldArr = fields.split(",");
        for (int i = 0; i < fieldArr.length; i++) {
            String fieldName = fieldArr[i];
            // 判断是否为主键
            if (fieldName.equals(pk)) {
                createSql.append(fieldName + " varchar primary key");
            } else {
                createSql.append(fieldName + " varchar");
            }
            if (i < fieldArr.length - 1) {
                createSql.append(",");
            }
        }
        createSql.append(")").append(ext);

        System.out.println("维度表的建表语句是：" + createSql);

        PreparedStatement ps = null;

        try {
            // 创建数据库操作对象
            ps = conn.prepareStatement(createSql.toString());
            // 执行SQL语句
            ps.execute();
        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("建表失败");
        }finally {
            // 释放资源
            if (ps != null) {
                try {
                    ps.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
